package com.example.wxg.stream.custom.source;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @author void
 * @date 2021/11/12 12:51
 * @desc
 */
public class MyRichParallelSource extends RichParallelSourceFunction<Long> {

    private Long count = 1L;
    private boolean isRunning = true;

    @Override
    public void run(SourceContext<Long> sourceContext) throws Exception {
        while (isRunning){
            sourceContext.collect(count);
            count++;
            //每秒产生一条数据
            Thread.sleep(1000L);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("open...");
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}

